package com.xiaochong.tir.api.component;

import com.xiaochong.tir.api.config.hbase.HbaseConnectionFactory;
import com.xiaochong.tir.common.exception.hbase.HbaseCloseExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseConnectionExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseGetTableExcetion;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by cx on 2018/4/24.
 */
@Component
public class HbaseBaseQueryComponent {


    @Autowired
    private HbaseConnectionFactory hbaseConnectionFactory;

    /**
     * 根据rowKey查询时间
     * @param tableName 表名
     * @param rowKey
     * @return
     * @throws HbaseConnectionExcetion
     * @throws HbaseGetTableExcetion
     * @throws IOException
     */
    public Long selectByTimestamp(String tableName,String rowKey) throws HbaseConnectionExcetion, HbaseGetTableExcetion, IOException, HbaseCloseExcetion {
        Long timestamp = null;
        Connection conn  = hbaseConnectionFactory.getConnection();
        Table table= hbaseConnectionFactory.getTable(conn,tableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = table.get(get);
        Cell cells[]= result.rawCells();
        if(null != cells && cells.length>0){
            timestamp =  cells[0].getTimestamp();
        }
        hbaseConnectionFactory.closeAll(conn,table);

        return timestamp;
    }

    /**
     * 读取hbase 数据
     * @param tableName
     * @param rowKey
     * @return param table ，rowKey
     * @throws HbaseConnectionExcetion
     * @throws HbaseGetTableExcetion
     * @throws IOException
     */
    public  Map<String,Object> selectData(String tableName,String rowKey) throws HbaseConnectionExcetion, HbaseGetTableExcetion, IOException, HbaseCloseExcetion {
        Map<String,Object> paramMap = new HashMap<String,Object>();
        Connection conn  = hbaseConnectionFactory.getConnection();
        Table table= hbaseConnectionFactory.getTable(conn,tableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = table.get(get);
        Cell cells[]= result.rawCells();
        for(Cell cell : cells){
            String qualifier = new String(CellUtil.cloneQualifier(cell),"UTF-8");
            String value = new String(CellUtil.cloneValue(cell),"UTF-8");
            if(StringUtils.isNotBlank(value) && !value.equals("null")){
                paramMap.put(qualifier,value);
            }

        }
        hbaseConnectionFactory.closeAll(conn,table);
        return paramMap;
    }

    public List<Long> seleceDateGetVersion(String tableName, String rowKey) throws HbaseCloseExcetion, IOException, HbaseGetTableExcetion, HbaseConnectionExcetion{
        List<Long> list = new ArrayList<>();
//        Connection conn  = hbaseConnectionFactory.getConnection();
//        Table table= hbaseConnectionFactory.getTable(conn,tableName);
//        Get get = new Get(Bytes.toBytes(rowKey));
//        get.setMaxVersions(10);
//        HbaseTableRFamilyBo familyBo = ApiServiceStatusCodeUtil.hbaseTbaleMap.get(tableName);
//        get.addColumn(Bytes.toBytes(familyBo.getFamily()),Bytes.toBytes(familyBo.getQualifier()));
//        Result result = table.get(get);
//        Cell cells[]= result.rawCells();
//        for(Cell cell : cells){
//
//            list.add(cell.getTimestamp());
//
//        }
//        closeHbase(conn,table,tableName);
        return list;
    }

    /**
     * 读取hbase 数据
     * @param tableName
     * @param rowKey
     * @return param key 为列名，value 为值 ,timeStamp
     * @throws HbaseConnectionExcetion
     * @throws HbaseGetTableExcetion
     * @throws IOException
     */
    public  Map<String,Object> selectDataByTimeStamp(String tableName,String rowKey,Long timeStamp) throws HbaseConnectionExcetion, HbaseGetTableExcetion, IOException, HbaseCloseExcetion {
        Map<String,Object> paramMap = new HashMap<String,Object>();
        Connection conn  = hbaseConnectionFactory.getConnection();
        Table table= hbaseConnectionFactory.getTable(conn,tableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        get.setTimeStamp(timeStamp);
        Result result = table.get(get);
        Cell cells[]= result.rawCells();
        for(Cell cell : cells){
            String qualifier = new String(CellUtil.cloneQualifier(cell),"UTF-8");
            String value = new String(CellUtil.cloneValue(cell),"UTF-8");
            if(StringUtils.isNotBlank(value) && !value.equals("null")){
                paramMap.put(qualifier,value);
            }

        }
        hbaseConnectionFactory.closeAll(conn,table);
        return paramMap;
    }

    /**
     * 指定列族列明查询
     * @param tableName
     * @param rowKey
     * @return param key 为列名，value 为值
     * @throws HbaseConnectionExcetion
     * @throws HbaseGetTableExcetion
     * @throws IOException
     */
    public  String selectByQualifierData(String tableName,String family,String qualifier ,String rowKey) throws HbaseConnectionExcetion, HbaseGetTableExcetion, IOException, HbaseCloseExcetion {
        Connection conn  = hbaseConnectionFactory.getConnection();
        Table table= hbaseConnectionFactory.getTable(conn,tableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(family),Bytes.toBytes(qualifier));
        Result result = table.get(get);
        String resultValue = null;
        Cell cells[]= result.rawCells();
        for(Cell cell : cells){
            resultValue = new String(CellUtil.cloneValue(cell),"UTF-8");
        }
        hbaseConnectionFactory.closeAll(conn,table);

        return resultValue;
    }
}
